#include "threadpool.h"

#include <functional>
#include <thread>
#include <iostream>
#include <chrono>

const int TASK_MAX_THRESHHOLD = INT32_MAX;
const int THREAD_MAX_THRESHHOLD = 10;
const int THREAD_MAX_IDLE_TIME = 60;

ThreadPool::ThreadPool() 
    : poolMode_(PoolMode::MODE_FIXED)
    , initThreadSize_(0)
    , taskSize_(0)
    , taskQueMaxThreshHold_(TASK_MAX_THRESHHOLD)
    , isRunning_(false)
    , idleThreadSize_(0)
    , threadSizeThreshHold_(THREAD_MAX_THRESHHOLD)
    , curThreadSize_(0)
{

}

ThreadPool::~ThreadPool()
{
    isRunning_ = false;

    // 等待线程池所有线程返回 阻塞 & 正在执行任务
    std::unique_lock<std::mutex> lock(taskMtx_);
    notEmpty_.notify_all();
    exit_.wait(lock, [&]()->bool {return threads_.size() == 0;});
}

void ThreadPool::setMode(PoolMode mode)
{
    if(checkRunningState())
        return;
    poolMode_ = mode;
}

// 设置任务上限
void ThreadPool::setTaskQueMaxThreshHold(int size) 
{
    if(checkRunningState())
        return;
    taskQueMaxThreshHold_ = size;
}

// 设置cached线程上限
void ThreadPool::setThreadSizeThreshHold(int size) 
{
    if(checkRunningState())
        return;
    if(poolMode_ == PoolMode::MODE_CACHED)
        threadSizeThreshHold_ = size;
}

// 开启线程池
void ThreadPool::start(size_t initThreadSize)
{
    isRunning_ = true;

    // 创建线程对象
    initThreadSize_ = initThreadSize;
    for(int i = 0; i < initThreadSize_; ++i) {
        auto thread = std::make_unique<Thread>(
            std::bind(&ThreadPool::threadFunction, this, std::placeholders::_1));
        int threadId = thread->getThreadId();
        threads_.emplace(threadId, std::move(thread));
        curThreadSize_++;
    }

    // 启动所有线程执行线程函数
    for(int i = 0; i < initThreadSize_; ++i) {
        threads_[i]->start();
        idleThreadSize_++;
    }

}
    
// 向线程池提交任务
Result ThreadPool::submitTask(std::shared_ptr<Task> task)
{
    std::unique_lock<std::mutex> lock(taskMtx_);
    // 提交任务超时
    if(!notFull_.wait_for(lock, std::chrono::seconds(1), 
        [&]()->bool { return taskQue_.size() < taskQueMaxThreshHold_;}))
    {
        std::cout << "taskQueue is full, submit Task fail..." << std::endl;
        // 不能使用返回值隶属于某个任务
        // 因为线程池执行完任务后就会将任务从任务队列中删除，此时task析构了，而result依赖于task
        // 因此不能这么设计，需要将task设置给某个result
        return Result(task, false);
    }

    taskQue_.emplace(task);
    taskSize_++;
    notEmpty_.notify_all();

    // cache模式， 在任务小而快的情景下根据任务数量和空闲线程数量判断是否创建新的线程
    if(poolMode_ == PoolMode::MODE_CACHED 
        && taskSize_ > idleThreadSize_ && curThreadSize_ < threadSizeThreshHold_) 
    {
        std::cout << " 线程被创建! " << std::endl;
        auto thread = std::make_unique<Thread>(
            std::bind(&ThreadPool::threadFunction, this, std::placeholders::_1));
        int threadId = thread->getThreadId();
        threads_.emplace(threadId, std::move(thread));

        // 启动线程并修改线程池状态
        threads_[threadId]->start();
        curThreadSize_++;
        idleThreadSize_++;
    }

    return Result(task);
}

void ThreadPool::threadFunction(int threadId)
{
    // 获取上次执行任务的时间
    auto lastTime = std::chrono::high_resolution_clock().now();

    // 线程不断获取任务并且执行
    while(true) 
    {
        // 获取任务
        std::shared_ptr<Task> task;
        {
            std::unique_lock<std::mutex> lock(taskMtx_);
            std::cout << " 线程 : " << std::this_thread::get_id() << " 尝试获取任务... " << std::endl;

            // 每秒返回一次，需要判断空闲时间
            while(taskQue_.size() == 0) 
            {
                // 线程池回收
                if(!isRunning_) 
                {
                    threads_.erase(threadId);
                    std::cout << " 线程 " << std::this_thread::get_id() << " 因线程池关闭被回收! " << std::endl;
                    exit_.notify_all();
                    return;
                }

                // cache模式下，如果创建了过多的线程，需要将多余的线程进行回收
                // 空闲60s以上的线程则对线程进行回收
                if(poolMode_ == PoolMode::MODE_CACHED) 
                {
                    // 条件变量超时返回的
                    if(std::cv_status::timeout == notEmpty_.wait_for(lock, std::chrono::seconds(1)))
                    {   
                        // 获取当前时间，计算差值是否超过60s
                        auto nowTime = std::chrono::high_resolution_clock().now();
                        auto durTime = std::chrono::duration_cast<std::chrono::seconds>(nowTime - lastTime);
                        // 超时回收线程
                        if(durTime.count() >= THREAD_MAX_IDLE_TIME && curThreadSize_ > initThreadSize_)
                        {
                            threads_.erase(threadId);
                            curThreadSize_--;
                            idleThreadSize_--;
                                
                            std::cout << " 线程 " << std::this_thread::get_id() << " 被回收! " << std::endl;
                            return;
                        }
                    }
                }
                else
                {
                    notEmpty_.wait(lock);
                }
                   
            }

            idleThreadSize_--;
            std::cout << " 线程 : " << std::this_thread::get_id() << " 获取任务成功... " << std::endl;
            task = taskQue_.front();
            taskQue_.pop();
            taskSize_--;

            if(taskQue_.size() > 0) 
                notEmpty_.notify_all();
            notFull_.notify_all();
        }

        // 执行任务，设置任务的返回值给到Result
        if(task != nullptr) {
            task->exec();
        }

        idleThreadSize_++;
        // 更新线程执行完任务的时间
        lastTime = std::chrono::high_resolution_clock().now();
    }
}

bool ThreadPool::checkRunningState() const
{
    return isRunning_;
}

int Thread::generateId_  = 0;

Thread::Thread(ThreadFunction func)
    : func_(func)
    , threadId_(generateId_++)
{

}

Thread::~Thread()
{

}

// 启动线程执行线程函数
void Thread::start() 
{
    std::thread t(func_, threadId_);
    t.detach();
}

int Thread::getThreadId() const
{
    return threadId_;
}

Semaphore::Semaphore(int limit)
    : resLimit_(limit)
    , isExit_(false)
{

}

Semaphore::~Semaphore()
{
    isExit_ = true;
}

// 获取一个信号量资源
void Semaphore::wait()
{
    if(isExit_) {
        return;
    }

    std::unique_lock<std::mutex> lock(mtx_);
    cond_.wait(lock, [&]()->bool {return resLimit_ > 0;});
    resLimit_--;
}

// 增加一个信号量资源
void Semaphore::post()
{   
    if(isExit_) {
        return;
    }

    std::unique_lock<std::mutex> lock(mtx_);
    resLimit_++;
    cond_.notify_all();
    /* 
        在notify_all上出现阻塞
        在vs下，条件变量析构会释放相应的资源，但是在linux没有释放相应资源
        导致notify_all状态失效
    */
}

Result::Result(std::shared_ptr<Task> task, bool isVaild)
    : task_(task)
    , isVaild_(isVaild)
{
    task_->setResult(this);
}

void Result::setVal(Any any)
{
    // 存储任务返回值
    this->any_ = std::move(any);
    sem_.post();
}


Any Result::get()
{
    if(!isVaild_) 
        return "";

    // task任务没有执行完成，则进行阻塞
    sem_.wait();
    return std::move(any_);
}

Task::Task()
    : result_(nullptr)
{

}

// 实现多态调用并设置返回值
void Task::exec() 
{   
    if(result_ != nullptr)
        result_->setVal(run());
}

void Task::setResult(Result *result)
{
    result_ = result;
}